#include "cRTPClient.h"
#include "com_net.h"
#include "com_debug.h"
#include <pthread.h>
#include <string.h>
#include <sys/time.h>
#include "Pcapfile.h"
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <arpa/inet.h>
#ifdef MEM_WATCH
#include"memwatch.h"
#endif

pthread_mutex_t info_lock;
char time_info[128];
cRTPClient::cRTPClient():
mbRead(false),
mbParse(false),
mbTime(false),
mfistSequenceNumber(0),
mlostCount(0),
serIp(NULL),
m_socket(-1),
m_port(0),
pQueNet(NULL),
pQueH264(NULL),
pRTPParse(NULL)
{
	memset(&mlastRtpInfo,0,sizeof(mlastRtpInfo));
}
cRTPClient::~cRTPClient()
{

}
int cRTPClient::rtp_init(const char* serip)
{
    serIp=strdup(serip);
//创建网络数据缓冲区，h264输出缓冲区，创建socket , 创建rtp读线程和rtp解析线程
	pQueNet = new CQueue(300,1500);
    pQueH264 = new CQueue(50,512*1024);
	

	m_socket = udp_net_setup();
    unsigned recvBUf = increaseReceiveBufferTo(m_socket,2000000);
    DEBUG_INFO1("rtp_UDP_recv_buf %u m_socket = %d\n",recvBUf, m_socket);
	return udp_get_port(m_socket, &m_port);
	
	#if 0
	pthread_t pth_read;
	if(pthread_create(&pth_read,NULL,&cRTPClient::read_thread,this))
	{
		DEBUG_ERR("read thread create err!\n");
	}

	pthread_t pth_parse;
	if(pthread_create(&pth_read,NULL,&cRTPClient::parse_thread,this))
	{
		DEBUG_ERR("parse thread create err!\n");
	}
	#endif

}
int cRTPClient::rtp_deinit()
{
//	先结束两个线程
	rtp_stop();
	show_stream_info();
	delete pRTPParse;

	closeSocket(m_socket);
	if(serIp)
	{
		free(serIp);
		serIp = NULL;
	}


	delete pQueNet;
	delete pQueH264;
	return RET_SUCESS;
}

int cRTPClient::rtp_start()
{
	int ret = RET_SUCESS;
	mbParse = true;
	mbRead = true;
	mbTime = true;
	//开始线程
	if(pthread_create(&mpth_read,NULL,&cRTPClient::read_thread,this))
	{
		ret = RET_ERR;
		DEBUG_ERR("read thread create err!\n");
	}
#if 0
	if(pthread_create(&mpth_time,NULL,&cRTPClient::time_thread,this))
	{
		DEBUG_ERR("time thread create err!\n");
	}
#endif
#if 1
	if(pthread_create(&mpth_parse,NULL,&cRTPClient::parse_thread,this))
	{
		ret = RET_ERR;
		DEBUG_ERR("parse thread create err!\n");
	}
#endif
	
	return ret;
}
int cRTPClient::rtp_stop()
{
	mbParse = false;
	mbRead = false;

	pthread_join(mpth_read,NULL);
	pthread_join(mpth_parse,NULL);
	pthread_join(mpth_time,NULL);
	return RET_SUCESS;
}
int cRTPClient::rtp_getUDPPort()

{
	return m_port;
}
int cRTPClient::rtp_UDPConnect(unsigned serPort)
{
	return udp_connect(m_socket, serIp, serPort);
}

int cRTPClient::rtp_requestFrameBuffer(unsigned char **pdata,int *plen)
{
	return pQueH264->getbuffer(pdata, plen);
}
int cRTPClient::rtp_releaseBuffer(int index)
{
	return pQueH264->releasebuffer(index);
}
int cRTPClient::rtp_getFrame(unsigned char *data, int len)
{
	return pQueH264->pop(data,len);
}
int cRTPClient::set_header_info(uint8_t* pHeader,RTP_PACK_HEADER_INFO* pHeaderInfo )
{
	if(pHeader == NULL || pHeaderInfo == NULL)
	{
		DEBUG_WARN("pram erro \n");
		return RET_ERR;
	}

	#if 0
	pHeaderInfo->version = pHeader[0] & (0x03);
	pHeaderInfo->marker = pHeader[1] & (0x01);
	pHeaderInfo->payloadType = pHeader[1] & ~(0x01);
	#endif
	pHeaderInfo->version = LEFT_CHAR_BIT_GET(pHeader,0,2);
	pHeaderInfo->marker = LEFT_CHAR_BIT_GET(pHeader+1,0,1);
	pHeaderInfo->payloadType = LEFT_CHAR_BIT_GET(pHeader+1,1,7);

	
	pHeaderInfo->sequenceNumber = LITTLE_MAKE_UNINT16(pHeader+2);
	pHeaderInfo->timeStamp = LITTLE_MAKE_UNINT32(pHeader+4);
	pHeaderInfo->ssrc = LITTLE_MAKE_UNINT32(pHeader+8); 
	//DEBUG_INFO1("sequenceNumber %u timeStamp %u ssrc %u payloadType %d \n",pHeaderInfo->sequenceNumber,pHeaderInfo->timeStamp,pHeaderInfo->ssrc,pHeaderInfo->payloadType );
	return RET_SUCESS;
}
void cRTPClient::show_stream_info()
{
	DEBUG_WARN("all: rtp frame had losted  %d, expect:%d \n",mlostCount>0?mlostCount:0,mlastRtpInfo.sequenceNumber-mfistSequenceNumber);
}

bool cRTPClient::check_header(RTP_PACK_HEADER_INFO* pHeaderInfo)
{
	if(mfistSequenceNumber == 0)
	{// the fist frame
		mfistSequenceNumber = pHeaderInfo->sequenceNumber;
		memcpy(&mlastRtpInfo, pHeaderInfo,sizeof(mlastRtpInfo));

		if(pHeaderInfo->payloadType == 96 && pRTPParse == NULL)
		{
		    DEBUG_INFO3("carete h264rtpParser \n");
			pRTPParse = new cH264RTP(pQueNet,pQueH264);
		}
		return true;
	}
	if(pHeaderInfo->ssrc == mlastRtpInfo.ssrc)
	{
		if(pHeaderInfo->payloadType == mlastRtpInfo.payloadType)
		{
			if(pHeaderInfo->sequenceNumber - mlastRtpInfo.sequenceNumber == 1)
			{
			}
			else
			{
				mlostCount += pHeaderInfo->sequenceNumber - 1 - mlastRtpInfo.sequenceNumber;
				DEBUG_WARN("all: rtp frame had losted mlostCount %d ,expect:%d fistseq:%u curseq: %u lastseq: %u\n",mlostCount,pHeaderInfo->sequenceNumber-mfistSequenceNumber,mfistSequenceNumber,pHeaderInfo->sequenceNumber,mlastRtpInfo.sequenceNumber);
			    if(pHeaderInfo->sequenceNumber < mlastRtpInfo.sequenceNumber)
                {
                    DEBUG_WARN("seq erro lastseq: %u curseq: %u \n",mlastRtpInfo.sequenceNumber,pHeaderInfo->sequenceNumber);
                }
			}
			memcpy(&mlastRtpInfo, pHeaderInfo,sizeof(mlastRtpInfo));
			return true;
		}
		else
		{
			DEBUG_INFO3("loadtype %d  lastloadtype %d\n",pHeaderInfo->payloadType,mlastRtpInfo.payloadType);
		}
	}
	else
	{	
		DEBUG_INFO3("SSRC erro pHeaderInfo->ssrc %u mlastRtpInfo.ssrc %u \n",pHeaderInfo->ssrc, mlastRtpInfo.ssrc);
	}
	
	return false;
}

int cRTPClient::parse()
{
	uint8_t *pData;
	int dataLen=0,index = -1;
	RTP_PACK_HEADER_INFO rtpInfo;
	index = pQueNet->getbuffer(&pData, &dataLen);
	if(index >= 0)
	{
		set_header_info(pData,&rtpInfo);
		if(check_header(&rtpInfo))
		{
		    if(pRTPParse)
		    {
                pRTPParse->parse_payload(pData + 12, dataLen - 12);
            } else{
		        DEBUG_INFO3("no parser\n");
		    }
		}
		else
		{
			DEBUG_ERR("head check erro\n");
		}
		pQueNet->releasebuffer(index);
	}

	return RET_SUCESS;
}
#if 1
void* cRTPClient::time_thread(void *pRtpclient)
{
	printf("time_thread start11\n");
	cRTPClient * pRtp = (cRTPClient *)pRtpclient;
	int time_sockfd, time_ret, send_socket_fd;
	struct sockaddr_in time_cli_addr, send_socket, connect_addr;
	char buf[200];
	int data =1;
	socklen_t addrlen = sizeof( struct sockaddr);
	pthread_mutex_init(&info_lock, NULL);
	time_sockfd = socket(AF_INET, SOCK_DGRAM , 0);
	if(time_sockfd < 0) {
		printf("failed to time_sockfd\n");
	}
	setsockopt(time_sockfd, SOL_SOCKET, MSG_NOSIGNAL, &data, sizeof (int));
	time_cli_addr.sin_family = AF_INET;
	time_cli_addr.sin_port   = htons(8754); //recv_port
	time_cli_addr.sin_addr.s_addr = INADDR_ANY;
	if( bind(time_sockfd, (struct sockaddr*)&time_cli_addr, sizeof(struct sockaddr_in)) < 0)
	{
		printf("Fail to bind time_sockfd\n");
	}

	send_socket_fd = socket(AF_INET, SOCK_DGRAM , 0);
	send_socket.sin_family = AF_INET;
	send_socket.sin_port = htons(8854); //send_port
	send_socket.sin_addr.s_addr = inet_addr("192.168.110.173");

	connect_addr.sin_port = htons(8854);
	connect_addr.sin_addr.s_addr = inet_addr("192.168.110.173");
	if(connect(time_sockfd,(struct sockaddr *)&connect_addr,sizeof(struct sockaddr_in)) < 0) {
		printf("connect failed\n");
	}
	sendto(time_sockfd, "helloaa", 7, 0, (struct sockaddr*)&send_socket, sizeof(send_socket));
	while(1) {
		//time_ret = sendto(time_sockfd, "helloaa", 7, 0, (struct sockaddr*)&send_socket, sizeof(send_socket));
		time_ret = recvfrom(time_sockfd, (char*)buf, 200, 0, (struct sockaddr*)&time_cli_addr, &addrlen);
		pthread_mutex_lock(&info_lock);
		memcpy(time_info, buf, 24);
		pthread_mutex_unlock(&info_lock);
		//printf("%s\n", time_info);
		usleep(500);
	}

}
#endif
#define FU_INDICATE 12
#define FU_HEAD 13
#define FU_A_F 28
//#define time_test
//#define dbg_test
//#define detail_p
void* cRTPClient::read_thread(void *pRtpclient)
{
	cRTPClient * pRtp = (cRTPClient *)pRtpclient;
	struct timeval tv1, tv2, tv3, tv4, tv5;
    struct tm *pLocalTime;
    time_t timep;
	unsigned long delay = 0;
	//DEBUG_INFO1("fread ------------\n\n");
#if 0
	FILE *fp_rtp_data = fopen("./fp_rtp_data","w+");
	char my_head[64]={0};
	printf("read thread 1\n");
	if(fp_rtp_data == NULL)
	{
		DEBUG_ERR("fopen erro \n");
	}
	setvbuf(fp_rtp_data,NULL,_IONBF,0);
	printf("read thread 2\n");
	char *buffer = (char *)malloc(2000);
	int ret =0;
	while(1)
	{
		printf("rPtp->m_socke = %d\n", pRtp->m_socket);
		ret = writeSocket(pRtp->m_socket, "hello", 5);
		ret = readSocket(pRtp->m_socket, buffer, 2000);
		//printf("ret = %d\n", ret);
		if(ret > 0 && fp_rtp_data)
		{
			snprintf(my_head,sizeof(my_head),"udp_data:%d",ret);
			fwrite(my_head,1,sizeof(my_head),fp_rtp_data);
			fwrite(buffer,1,ret,fp_rtp_data);
		}
	}
	fclose(fp_rtp_data);
#else
	#if 1
		int ret =0;
		uint8_t *buffer = (uint8_t *)malloc(1500);
		//FILE *fp_rtp_data = fopen("./fp_rtp_data.txt","w+");
		//setvbuf(fp_rtp_data,NULL,_IONBF,0);
		ret = writeSocket(pRtp->m_socket, "hello", 5);
		while(pRtp->mbRead)
		{
			//printf("pRtp->mbRead read\n");
			gettimeofday(&tv1, NULL);
			ret = readSocket(pRtp->m_socket, (char *)buffer, 2000);
			#if 0
			pthread_mutex_lock(&info_lock);
			printf("%s\n", time_info);
			pthread_mutex_unlock(&info_lock);
			#endif
			#ifdef dbg_test
			for(int i = 0; i <= ret/16; i++) {
				for(int j = 0; j < 16; j++) {
					printf("%02x ", buffer[i*16+j]);
				}
				printf("\n");
			}
			#endif
			#ifdef time_test
			//NAL 单包时间
			//printf("buffer[FU_INDICATE] & 0x1f = %d\n", buffer[FU_INDICATE] & 0x1f);
			printf("sequence = %ld\n", buffer[2]<<8|buffer[3]);
			if(((buffer[FU_INDICATE] & 0x1f) < 24) && ((buffer[FU_INDICATE] & 0x1f) > 0)) //buffer[FU_INDICATE] & 0x1f == FU TYPE
			{
				gettimeofday(&tv2, NULL);
				delay += ((tv2.tv_usec -tv1.tv_usec) / 1) + 1000000*(tv2.tv_sec - tv1.tv_sec);
				#ifdef detail_p
				printf("NAL UNIT PACKET use time %ld us\n", ((tv2.tv_usec -tv1.tv_usec) / 1) + 1000000*(tv2.tv_sec - tv1.tv_sec));
				#endif
			} else if((buffer[FU_INDICATE] & 0x1f) == FU_A_F) {
				//printf("NAL FU-A\n");
				if(buffer[FU_HEAD] & 0x80)  //分片第一包
				{
					gettimeofday(&tv3, NULL);
					delay += ((tv3.tv_usec -tv1.tv_usec) / 1) + 1000000*(tv3.tv_sec - tv1.tv_sec);
					#ifdef detail_p
					printf("NAL FU-A start use time %ld us\n", ((tv3.tv_usec -tv1.tv_usec) / 1) + 1000000*(tv3.tv_sec - tv1.tv_sec));
					#endif
				} else if(buffer[FU_HEAD] & 0x40) { //分片最后一包
					gettimeofday(&tv4, NULL);
					delay += ((tv4.tv_usec -tv1.tv_usec) / 1) + 1000000*(tv4.tv_sec - tv1.tv_sec);
					printf("recv one frame use %ld us\n", delay);
					delay = 0;
					#if 0
					pthread_mutex_lock(&info_lock);
					printf("%s\n", time_info);
					pthread_mutex_unlock(&info_lock);
					#endif
					#ifdef detail_p
					printf("NAL FU-A end use time %ld us\n", ((tv4.tv_usec -tv1.tv_usec) / 1) + 1000000*(tv4.tv_sec - tv1.tv_sec));
					#endif
				} else {  //分片中的包
					gettimeofday(&tv5, NULL);
					delay += ((tv5.tv_usec -tv1.tv_usec) / 1) + 1000000*(tv5.tv_sec - tv1.tv_sec);
					#ifdef detail_p
					printf("process FU-A use time %ld us\n", ((tv5.tv_usec -tv1.tv_usec) / 1) + 1000000*(tv5.tv_sec - tv1.tv_sec));
					#endif
				}
			}

			#endif
			//fwrite(buffer,1,ret,fp_rtp_data);
			if(ret >0)
			{
				//printf("net push que\n");
				pRtp->pQueNet->push(buffer, ret);
			}
		}
		//fclose(fp_rtp_data);
		free(buffer);
	#elif 0
		int ret =0;
		uint8_t buffer[1024*20]={0};
		CpcapPare mpcap;
		mpcap.init("my_rtsp.pcap",0);
		while((ret = mpcap.getFrames(buffer,sizeof(buffer)))>0)
		{
			pRtp->pQueNet->push(buffer, ret);
		}
	#else
		int ret =0,datalen = 0;
		uint8_t head[64]={0};
		uint8_t *buffer = (uint8_t*)malloc(1024*30);
		FILE *fp_test = fopen("fp_rtp_data","r");
		if(fp_test == NULL)
		{
			DEBUG_ERR("erro to open file \n");
		}

		while(1)
		{
			memset(head,0,sizeof(head));
			ret = fread(head,1,sizeof(head),fp_test);
			DEBUG_INFO1("fread %s\n",head);
			usleep(1000000/50);
			if(ret == sizeof(head) )
			{
				if(sscanf((const char*)head, "udp_data:%d", &datalen) != 1)
				{
					DEBUG_ERR("scan erro:head:%s\n",head);
					break;
				}
				
				ret = fread(buffer,1,datalen,fp_test);
				if(ret < datalen)
				{
					DEBUG_INFO1("read erro: ret %d\n",ret);
					break;
				}
				else
				{
					pRtp->pQueNet->push(buffer, ret);
				}
				
			}
			else
			{
				DEBUG_INFO1("read erro: ret %d\n",ret);
				break;
			}
		}

		fclose(fp_test);
		free(buffer);
		
	#endif
#endif
		return NULL;
}
void* cRTPClient::parse_thread(void *pRtpclient)
{
	cRTPClient * pRtp = (cRTPClient *)pRtpclient;
	while(pRtp->mbParse)
	{
		pRtp->parse();
	}
	return NULL;
}



